package com.etc;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * @Author kalista
 * @Description kafka 消息生产者
 * @Date 2021/5/22  14:54
 **/
public class Producer extends Thread {

    private final static Logger log = LoggerFactory.getLogger(Producer.class);


    private final boolean isAsync;
    private final String topic;
    private final KafkaProducer<Integer, String> producer;

    public Producer(String topic, boolean isAsync) {
        Properties properties = new Properties();
        //kafka地址
        properties.put("bootstrap.servers", "hdp-01:9092,hdp-02:9092,hdp-03:9092");
        // kafka key 序列化
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // keafka value 序列化
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer<>(properties);
        this.topic = topic;
        this.isAsync = isAsync;

    }


    @Override
    public void run() {
        int meesageNo = 1;
        while (true) {
            // message_1
            String message = "hello,wcc";
            log.warn("message:" + message);
            long startTime = System.currentTimeMillis();
            // 异步发送
            producer.send(new ProducerRecord<>(topic, message));
            log.info("send message:={}", message);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            meesageNo++;
        }
    }


    public static void main(String[] args) {
        Producer producer = new Producer("my-topic",false);
        producer.start();
    }
}
